"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : 使用jdbc创建复制槽命名规范验证
Create At   : 2022/03/29
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.修改逻辑复制相关参数
    4.创建数据库，创建用户，配置白名单
    5.jdbc准备
    6.创建复制槽，命名符合规范
    7.创建复制槽，命名不符合规范
    8.创建已存在的slotname
    9.创建复制槽，命名超过长度
    10.删除逻辑复制槽
    11.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.修改逻辑复制相关参数成功
    4.创建成功，配置白名单成功
    5.jdbc准备成功
    6.创建成功
    7.创建失败
    8.创建失败
    9.创建失败
    10.删除逻辑复制槽成功
    11.资源清理成功，恢复参数成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_root = Node('PrimaryRoot')
        self.standby_node = Node('Standby1DbUser')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0018')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.db_name = 'db_replication_0018'
        self.user_name = 'u_logic_replication_0018'
        self.slot_name_normal = 'slot_standby_logic_replication_0018_normal'
        self.slot_name_upper = 'SLOT_STANDBY_logic_replication_0018_upper'
        self.slot_name_63chars = 'slot_standby_logic_replicat_' \
                                 'slot_standby_logic_replication_0018'
        self.slot_name_unexp = 'slot_standby_logic_replication_0018_*'
        self.slot_name_64chars = 'slot_standby_logic_replicat__' \
                                 'slot_standby_logic_replication_0018'
        self.jdbc_path = os.path.join(self.dir_replication, 'libs')
        self.CLASSPATH_PATH = os.path.join(self.dir_replication, 'CLASSPATH')
        self.jdbc_log_path = os.path.join(self.dir_replication, 'jdbc.log')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_pri_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_pri_msg)
        self.assertEqual("OK", cp_pri_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

        text = "-----step4:创建数据库,创建用户,配置白名单;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop database if exists {self.db_name};
                    create database {self.db_name} template template0 \
                    encoding 'UTF8' lc_ctype 'en_US.UTF-8' \
                    lc_collate 'en_US.UTF-8';
                    drop user if exists {self.user_name};
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}';'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication {self.user_name}  sha256'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host replication {self.user_name} ' \
            f'{self.primary_node.db_host}/32 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step5:jdbc准备;expect:成功-----"
        self.logger.info(text)
        shell_cmd = f"mkdir {self.jdbc_path} && " \
            f"cd {macro.DB_SCRIPT_PATH}/../ && " \
            f"tar -xf `ls | grep -i jdbc | grep tar.gz ` -C {self.jdbc_path} " \
            f"&& chmod -R 777 {self.jdbc_path} && echo OK"
        self.logger.info(shell_cmd)
        shell_msg = self.primary_root.sh(shell_cmd).result()
        self.assertIn("OK", shell_msg, '执行失败' + text)

        self.logger.info("-----生成ClASSPATH-----")
        self.common.scp_file(self.primary_node, 'replication_tools.sh',
                             self.dir_replication)
        self.common.scp_file(self.primary_node, 'postgresql_right.jar',
                             self.dir_replication)
        self.tools_file_path = os.path.join(self.dir_replication,
                                            'replication_tools.sh')
        create_cmd = f"sh {self.tools_file_path} " \
            f"genClassPath {self.jdbc_path} > {self.CLASSPATH_PATH} && echo OK"
        self.logger.info(create_cmd)
        create_msg = self.primary_root.sh(create_cmd).result()
        self.assertIn("OK", create_msg, '执行失败' + text)

        self.logger.info("-----修改配置-----")
        self.common.scp_file(self.primary_node, 'conn.conf',
                             self.dir_replication)
        self.common.scp_file(self.primary_node, 'conn_rep1.conf',
                             self.dir_replication)
        self.common.scp_file(self.primary_node, 'conn_rep2.conf',
                             self.dir_replication)
        self.conf_files_path = os.path.join(self.dir_replication, 'conn*.conf')
        self.conf_1_path = os.path.join(self.dir_replication, 'conn_rep1.conf')
        self.conf_2_path = os.path.join(self.dir_replication, 'conn_rep2.conf')
        sed_cmd = f'''chmod 777 {self.dir_replication} && 
                           sed -i -e '/port=/c port={self.primary_node.db_port}' \
                           -e '/user=/c user={self.user_name}' \
                           -e "/password=/c password={macro.COMMON_PASSWD}" \
                           -e "/hostname=/c hostname={self.primary_node.db_host}" \
                           -e '/loggerfile=/c loggerfile={self.jdbc_log_path}' \
                           -e '/dbname=/c dbname={self.db_name}' \
                           -e '/loggerlevel=/c loggerlevel=error' \
                           {self.conf_files_path}'''
        self.logger.info(sed_cmd)
        sed_msg = self.primary_root.sh(sed_cmd).result()
        self.logger.info(sed_msg)
        self.assertNotIn("ERROR", sed_msg, '执行失败' + text)
        sed_cmd = f'''sed -i  -e \
                           "/hostname=/c hostname={self.standby_node.db_host}" \
                           -e '/port=/c port={self.standby_node.db_port}' \
                           {self.conf_2_path};'''
        self.logger.info(sed_cmd)
        sed_msg = self.primary_root.sh(sed_cmd).result()
        self.logger.info(sed_msg)
        self.assertNotIn("ERROR", sed_msg, '执行失败' + text)

    def test_standby_logicreplication(self):
        text = "-----step6:创建复制槽，命名符合规范;expect:创建成功-----"
        self.logger.info(text)
        self.common.scp_file(self.primary_node,
                             'Logical_replication_puls.class',
                             self.dir_replication)
        self.common.scp_file(self.primary_node,
                             'Logical_replication_puls.java',
                             self.dir_replication)
        names = [self.slot_name_normal, self.slot_name_upper,
                 self.slot_name_63chars]
        for name in names:
            create_cmd = f"source {macro.DB_ENV_PATH}; " \
                f"cd {self.dir_replication}; " \
                f"java -cp `cat CLASSPATH` Logical_replication_puls " \
                f"-F {self.conf_1_path} -M create_slot " \
                f"--args \"slotname={name}\""
            self.logger.info(create_cmd)
            create_msg = self.primary_node.sh(create_cmd).result()
            self.logger.info(create_msg)
            text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
            self.logger.info(text)
            sql_cmd = f"select * from pg_get_replication_slots();"
            sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
            self.logger.info(sql_msg)
            if name == self.slot_name_upper:
                name = name.lower()
            self.assertIn(name, sql_msg, '执行失败:' + text)

        text = "-----step7:创建复制槽，命名不符合规范;expect:创建失败-----"
        self.logger.info(text)
        create_cmd = f"source {macro.DB_ENV_PATH}; " \
            f"cd {self.dir_replication}; " \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_1_path} -M create_slot " \
            f"--args \"slotname={self.slot_name_unexp}\""
        self.logger.info(create_cmd)
        create_msg = self.primary_node.sh(create_cmd).result()
        self.logger.info(create_msg)
        self.assertIn("FATAL: replication command parser returned 1",
                      create_msg, '执行失败:' + text)

        text = "-----step8:创建已存在的slotname;expect:创建失败-----"
        self.logger.info(text)
        create_cmd = f"source {macro.DB_ENV_PATH}; " \
            f"cd {self.dir_replication}; " \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_1_path} -M create_slot " \
            f"--args \"slotname={self.slot_name_normal}\""
        self.logger.info(create_cmd)
        create_msg = self.primary_node.sh(create_cmd).result()
        self.logger.info(create_msg)
        self.assertIn(f"FATAL: replication slot \"{self.slot_name_normal}\" "
                      f"already exists", create_msg, '执行失败:' + text)

        text = "-----step9:创建复制槽，命名超过长度;expect:创建失败-----"
        self.logger.info(text)
        create_cmd = f"source {macro.DB_ENV_PATH}; " \
            f"cd {self.dir_replication}; " \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_1_path} -M create_slot " \
            f"--args \"slotname={self.slot_name_64chars}\""
        self.logger.info(create_cmd)
        create_msg = self.primary_node.sh(create_cmd).result()
        self.logger.info(create_msg)
        text = "-----验证创建逻辑复制槽失败;expect:创建失败-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertNotIn(self.slot_name_64chars, create_msg, '执行失败:' + text)

    def tearDown(self):
        text = "-----step10:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        names = [self.slot_name_normal, self.slot_name_upper,
                 self.slot_name_63chars]
        for name in names:
            drop_slot_cmd = f"source {macro.DB_ENV_PATH};" \
                f"cd {self.dir_replication};" \
                f"java -cp `cat CLASSPATH` Logical_replication_puls " \
                f"-F {self.conf_1_path} -M drop_slot " \
                f"--args \"slotname={name}\""
            self.logger.info(drop_slot_cmd)
            drop_slot_msg = self.primary_node.sh(drop_slot_cmd).result()
            self.logger.info(drop_slot_msg)
        self.logger.info("-----验证是否删除成功;expect:成功-----")
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)

        text = "-----step11:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step11.1:删除数据库，删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop database {self.db_name};
            drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step11.2:恢复postgresql.conf,pg_hba.conf文件，" \
                "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]];
            then rm -rf {self.postgresql_path};
            mv {self.postgresql_bak_path} {self.postgresql_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)

        text_3 = "-----step11.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.logger.info(del_cmd)
        self.primary_node.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        self.logger.info(primy_file_msg)
        self.assertNotIn(self.slot_name_normal and
                         self.slot_name_upper.lower() and
                        self.slot_name_63chars, sql_msg, '执行失败' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertEqual('OK', restore_primary_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_primary_2, '执行失败' + text_2)
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
